import logging
from typing import Any, Dict, List, Optional, Tuple

from calc_manager.plugins.plugins_manager import PluginManager
from calc_manager.base.runnable import Runnable
from calc_manager.operator import Operator, OperatorInfo
from calc_manager.operator.repo import OperatorRepository
from calc_manager.base.fields import FieldInfo
from .types import ParameterMapping
from pydantic import BaseModel
from calc_manager.plugins.plugin_object import PluginHelper

from .graph import AlgoGraph, OperatorComponent
from .param import ParameterMappingService


logger = logging.getLogger(__name__)

class AlgorithmInfo(BaseModel):
    """
    Represents information about an algorithm.

    Attributes:
        algo_name (str): The name of the algorithm.
        algo_desc (Optional[str]): An optional description of the algorithm. Defaults to None.
        graph_json (str): A JSON representation of the algorithm's graph structure.
    """

    algo_name: str
    algo_desc: Optional[str] = None
    graph_json: str


class OperatorComponentToOperatorInfoAdapter:
    """
    Adapter class to convert OperatorComponent to OperatorInfo.

    This class acts as a bridge between OperatorComponent and OperatorInfo,
    using an OperatorRepository to fetch the necessary information.

    Attributes:
        repository (OperatorRepository): The repository used to fetch operator information.
    """

    def __init__(self, repository: OperatorRepository):
        """
        Initialize the adapter with an OperatorRepository.

        Args:
            repository (OperatorRepository): The repository to use for fetching operator information.
        """
        self.repository = repository

    def get_operator_info(self, op_id: str) -> OperatorInfo:
        """
        Retrieve OperatorInfo for a given operator ID.

        This method fetches the operator information from the repository
        and returns it as an OperatorInfo object.

        Args:
            op_id (str): The ID of the operator to retrieve information for.

        Returns:
            OperatorInfo: The information about the requested operator.

        Raises:
            ValueError: If the operator information is not found for the given ID.
        """
        op = self.repository.get_op_by_id(op_id)
        if op is None:
            raise ValueError(f"Operator info not found for {op_id}")
        return op


class Algorithm:
    def __init__(
        self,
        algo_info: AlgorithmInfo,
        repo_adapt: OperatorComponentToOperatorInfoAdapter,
        param_mapping: ParameterMappingService,
        graph: AlgoGraph,
        **algo_params,
    ):
        """
        Initialize the Algorithm instance.

        Args:
            algo_info (AlgorithmInfo): Information about the algorithm.
            repo_adapt (OperatorComponentToOperatorInfoAdapter): Adapter for converting OperatorComponent to OperatorInfo.
            param_mapping (ParameterMappingService): Service for parameter mapping.
            graph (AlgoGraph): Graph representation of the algorithm.
            **algo_params: Additional algorithm parameters.

        Attributes:
            info (AlgorithmInfo): Stores the algorithm information.
            plugin_manager (None): Placeholder for plugin manager, initially set to None.
            graph (AlgoGraph): The graph representation of the algorithm.
            mappings (ParameterMappingService): Service for parameter mapping.
            algo_params (dict): Additional algorithm parameters.
            repo_adapt (OperatorComponentToOperatorInfoAdapter): Adapter for OperatorComponent to OperatorInfo conversion.
            operator_infos_by_id (Dict[str, OperatorInfo]): Dictionary of OperatorInfo objects indexed by their IDs.
            graph_operator_component_order (List): Ordered list of operator components in the graph.
            operation_info_exectution_order_by_graphid (List[Tuple[str, OperatorInfo]]): Ordered list of tuples containing graph IDs and corresponding OperatorInfo objects.
        """
        self.info = algo_info
        self.plugin_manager: Optional[PluginManager] = None
        self.graph = graph
        self.mappings = param_mapping
        self.algo_params = algo_params
        self.repo_adapt = repo_adapt
        self.operator_infos_by_id: Dict[str, OperatorInfo] = (self.get_all_operation_info_by_id())
        self.operator_infos_by_graph_id: Dict[str, OperatorInfo] = (self.get_all_operation_info_by_graph_id())
        self.graph_operator_component_order = self.graph.get_execution_order(
            self.graph.start_node
        )
        self.operation_info_exectution_order_by_graphid = self._get_operation_info_exection_order_by_graphid()

    def algo_run_task(self):
        logger.info("Starting algorithm run task for algorithm '%s' (id: %s)", self.info.algo_name, self.info.algo_desc)
        running_ctx = {}
        ops = self.get_operation_info_exection_order_by_graphid()
        op_dict = self.get_all_operators_by_id()

        # init all plugins that operator needed
        plugin_class_name_to_obj = {}
        results = {}
        plugin_helpers = {}
        if Operator.INVOKE_PLUGINS_KEY in self.algo_params:
            plugin_argv = self.algo_params[Operator.INVOKE_PLUGINS_KEY]

            plugin_helpers = Operator.get_plugin_objects_from_argv(self.plugin_manager, plugin_argv)
            plugins_obj = {k: v.get_plugin_ojbect() for k, v in plugin_helpers.items()}
            plugin_class_name_to_obj.update(plugins_obj)

        def pre_run(ctx, op: Operator, op_graph_id: str, invoke_argv, invoke_plugins):
            logger.info("Preparing to run operator '%s' (id: %s) with graph_id: %s", op_info.operator_name, op_info.operator_id, op_graph_id)
            argv = self.algo_params
            # do not dispos plugins because current running in one thread
            #extra_invoke_params = self._mapping_operator_plugins_to_extra_param(ctx, op_graph_id, argv, op)
            extra_invoke_params = op.get_plugin_objects_from_cls_name_obj_dict(plugin_class_name_to_obj)
            invoke_plugins.update(extra_invoke_params)
            logger.debug("Extra invoke params keys: %s", list(extra_invoke_params.keys()))

            #pop out raw plugin info
            argv.pop(Operator.INVOKE_PLUGINS_KEY, None)

            #mapping params to ctx
            self._mapping_param_to_ctx_argv(ctx, op_graph_id, argv)

            # mix argv and extra_invoke_params and store in the passed invoke_argv dict
            invoke_argv.update({**argv, **extra_invoke_params})

        def execute(op: Operator, invoke_argv):
            return op.invoke_func(**invoke_argv)
        
        def post_run(ctx, runnable_graph_id: str, ret):
            self._mapping_result_to_ctx(ctx, runnable_graph_id, ret)
            # Only print keys from ctx, including nested keys
            def get_nested_keys(d, prefix=''):
                keys = []
                for k, v in d.items():
                    new_key = f"{prefix}.{k}" if prefix else k
                    keys.append(new_key)
                    if isinstance(v, dict):
                        keys.extend(get_nested_keys(v, new_key))
                return keys
                
            ctx_keys = get_nested_keys(ctx)
            logger.debug(f"post_run ctx keys: {ctx_keys}")

        invoke_plugins = {}
        try:
            for op_graph_id, op_info in ops:
                op = op_dict[op_info.operator_id]
                invoke_argv = {}
                pre_run(running_ctx, op, op_graph_id, invoke_argv, invoke_plugins)
                ret = execute(op, invoke_argv)

                if isinstance(ret, BaseModel):
                    ret = ret.model_dump()
                    results.update(ret)
                else:
                    raise ValueError(f"Result of operator {op_info.operator_name} is not a pydantic BaseModel, type is {type(ret)}")

                post_run(running_ctx, op_graph_id, ret)
                #get op runnable and run
            return results
        except Exception as err:
            logger.error(f"Failed to run algorithm {self.info.algo_name}: {err}")
            raise err
        finally:
            #dispose plugins
            for k,v in plugin_helpers.items():
                logger.debug(f"dispose plugin {k}")
                v.dispose()

            #dispose operator
            for op_id, op in op_dict.items():
                logger.debug(f"dispose operator {op.info}")
                op.dispose()

    
    def set_plugin_manager(self, pm: PluginManager):
        """
        Set the plugin manager for the algorithm.

        Args:
            pm (PluginManager): The plugin manager to be set.
        """
        self.plugin_manager = pm

    def get_operation_info_exection_order_by_graphid(
        self,
    ) -> List[Tuple[str, OperatorInfo]]:
        """
        Get the execution order of operation information by graph ID.

        Returns:
            List[Tuple[str, OperatorInfo]]: A list of tuples containing graph IDs and corresponding OperatorInfo objects.
        """
        return self.operation_info_exectution_order_by_graphid

    def _get_operation_info_exection_order_by_graphid(
        self,
    ) -> List[Tuple[str, OperatorInfo]]:
        """
        Generate the execution order of operation information by graph ID.

        Returns:
            List[Tuple[str, OperatorInfo]]: A list of tuples containing graph IDs and corresponding OperatorInfo objects.
        """
        ret = []
        #TODO: optimize: use cache 
        for oc in self.graph.get_execution_order(self.graph.start_node):
            op = self.repo_adapt.get_operator_info(oc.component_id)
            ret.append((oc.component_graph_id, op))
        return ret

    #TODO: optimize: use cache
    def get_all_operation_info_by_graph_id(self) -> Dict[str, OperatorInfo]:
        """
        Get all operation information indexed by their IDs.

        Returns:
            Dict[str, OperatorInfo]: A dictionary of OperatorInfo objects indexed by their component IDs.
        """
        ret = {}
        for oc in self.graph.get_execution_order(self.graph.start_node):
            ret[oc.component_graph_id] = self.repo_adapt.get_operator_info(
                oc.component_id
            )
        return ret

    #TODO: optimize: use cache
    def get_all_operation_info_by_id(self) -> Dict[str, OperatorInfo]:
        """
        Get all operation information indexed by their IDs.

        Returns:
            Dict[str, OperatorInfo]: A dictionary of OperatorInfo objects indexed by their component IDs.
        """
        ret = {}
        for oc in self.graph.get_execution_order(self.graph.start_node):
            if oc.component_id not in ret:
                ret[oc.component_id] = self.repo_adapt.get_operator_info(
                    oc.component_id
                )
        return ret

    def get_all_operators_by_id(self) -> Dict[str, Operator]:
        """
        Get all operators indexed by their IDs.

        Returns:
            Dict[str, Operator]: A dictionary of Operator objects indexed by their IDs.
        """
        return {op_id: Operator(oi) for op_id, oi in self.operator_infos_by_id.items()}

    def _get_all_algo_plugin_invoke_params(self, op_dict: Dict[str, Operator]) -> Dict[str, List[FieldInfo]]:
        plugin_invoke_params_with_plugin_name: Dict[str, List[FieldInfo]] = {}
        if self.plugin_manager is None:
            return plugin_invoke_params_with_plugin_name
            
        for op_id, op in op_dict.items():
            plugin_invoke_params = op.get_operator_plugin_invoke_params(self.plugin_manager)
            logger.debug("Operator %s uses plugins: %s", op.info.operator_name, list(plugin_invoke_params.keys()))
            plugin_invoke_params_with_plugin_name.update(plugin_invoke_params)
        return plugin_invoke_params_with_plugin_name

    #TODO: use classmethod
    def _get_all_params(self, op_dict: Dict[str, Operator]) -> Dict[str, List[FieldInfo]]:
        """
        Get all parameters for each operator.

        Args:
            op_dict (Dict[str, Operator]): A dictionary of operators indexed by their IDs.

        Returns:
            Dict[str, List[FieldInfo]]: A dictionary where keys are operator IDs and values are lists of FieldInfo objects representing parameters.
        """
        ret = {}
        for op_id, op in op_dict.items():
            params_dict = op.extract_params()
            ret[op_id] = list(params_dict.values())
        return ret

    #TODO: use classmethod
    def _get_all_result(self, op_dict: Dict[str, Operator]) -> Dict[str, List[FieldInfo]]:
        """
        Get all results for each operator.

        Args:
            op_dict (Dict[str, Operator]): A dictionary of operators indexed by their IDs.

        Returns:
            Dict[str, List[FieldInfo]]: A dictionary where keys are operator IDs and values are lists of FieldInfo objects representing results.
        """
        ret = {}
        for op_id, op in op_dict.items():
            params_dict = op.extract_result()
            ret[op_id] = list(params_dict.values())
        return ret

    #TODO: use classmethod
    def get_all_params_results(self, op_dict: Dict[str, Operator]) -> Dict[str, Dict[str, List[FieldInfo]]]:
        """
        Get all parameters and results for each operator.

        Args:
            op_dict (Dict[str, Operator]): A dictionary of operators indexed by their IDs.

        Returns:
            Dict[str, Dict[str, List[FieldInfo]]]: A dictionary with 'params' and 'results' keys, each containing a dictionary of operator IDs mapped to lists of FieldInfo objects.
        """
        return {
            "params": self._get_all_params(op_dict),
            "results": self._get_all_result(op_dict)
        }

    #TODO: use classmethod
    def _get_param_info_from_op_id(self, op_id: str, type_name: str, op_dict: Dict[str, Operator]) -> FieldInfo:
        """
        Get parameter information for a specific operator and parameter type.

        Args:
            op_id (str): The ID of the operator.
            type_name (str): The name of the parameter type.
            op_dict (Dict[str, Operator]): A dictionary of operators indexed by their IDs.

        Returns:
            FieldInfo: The FieldInfo object for the specified parameter.
        """
        op = op_dict[op_id]
        params = op.extract_params()
        return params[type_name]

    #TODO: use classmethod
    def _get_res_info_from_op_id(self, op_id: str, type_name: str, op_dict: Dict[str, Operator]) -> FieldInfo:
        """
        Get result information for a specific operator and result type.

        Args:
            op_id (str): The ID of the operator.
            type_name (str): The name of the result type.
            op_dict (Dict[str, Operator]): A dictionary of operators indexed by their IDs.

        Returns:
            FieldInfo: The FieldInfo object for the specified result.
        """
        op = op_dict[op_id]
        results = op.extract_result()
        return results[type_name]

    @classmethod
    def _check_and_add_param_to_dict(cls, src: Dict[str, FieldInfo], fields: List[FieldInfo], check_type_conflict: bool = False):
        for field in fields:
            if field.name in src:
                if check_type_conflict:
                    if src[field.name].type != field.type:
                        raise ValueError(f"Field name '{field.name}' has different types: '{src[field.name].type}' and '{field.type}'")
                #if required need update
                if field.optional_field == "required":
                    logger.debug("Update required field %s to dict", field)
                    src[field.name] = field
            else:
                logger.debug("Add field %s to dict", field)
                src[field.name] = field

    @classmethod
    def _extend_params_list(cls, orig: List[FieldInfo], extend: List[FieldInfo], check_type_conflict: bool = False) -> List[FieldInfo]:
        #ret = orig.copy()
        ret = orig
        for field in extend:
            existing_field = next((f for f in ret if f.name == field.name), None)
            if existing_field:
                if check_type_conflict and existing_field.type != field.type:
                    raise ValueError(f"Field name '{field.name}' has different types: '{existing_field.type}' and '{field.type}'")
                #if required need update
                if field.optional_field == "required":
                    ret.remove(existing_field)
                    ret.append(field)
            else:
                ret.append(field)
        return ret

    @classmethod
    def _get_param_fields_without_mapped_param(cls, params_list: List[FieldInfo], param_filters: List[ParameterMapping]) -> List[FieldInfo]:
        mapped_params = {mapping.dst_param_name for mapping in param_filters}
        return [param for param in params_list if param.name not in mapped_params]

    def get_algo_plugin_params(self, op_dict: Dict[str, Operator]) -> Dict[str, List[FieldInfo]]:
        return self._get_all_algo_plugin_invoke_params(op_dict)

    def get_algo_input_params_with_mapping_filter(self, op_dict: Dict[str, Operator], check_type_conflict: bool = False) -> List[FieldInfo]:
        logger.debug("get_algo_input_params_with_mapping_filter start, check_type_conflict: %s", check_type_conflict)
        ret: Dict[str, FieldInfo] = {}
        params_dict = self._get_all_params(op_dict)
        for component in self.graph_operator_component_order:
            graph_id_com = component.component_graph_id
            upstram_component = self.graph.find_upstream_nodes(graph_id_com)
            # put all params
            if not upstram_component:
                fields_component = params_dict[component.component_id] 
                Algorithm._check_and_add_param_to_dict(ret, fields_component, check_type_conflict)
            else:
                params: List[FieldInfo] = []
                #extend param
                [Algorithm._extend_params_list(params, params_dict[component.component_id], check_type_conflict) for uc in upstram_component]
                filtered_params = Algorithm._get_param_fields_without_mapped_param(params, self.graph.get_upstream_nodes_mappings(component.component_graph_id))
                Algorithm._check_and_add_param_to_dict(ret, filtered_params, check_type_conflict)
        return list(ret.values())


    def check_mapping_filter_types(self, op_dict: Dict[str, Operator]):
        mappings = self.graph.get_all_node_mappings()
        for m in mappings:
            srctype = self._get_res_info_from_op_id(m.source_operator_id, m.src_param_name, op_dict)
            dsttype = self._get_param_info_from_op_id(m.dst_operator_id, m.dst_param_name, op_dict)
            if srctype.type != dsttype.type:
                raise Exception("Mapping filter types are not correct: %s-%s[%s] != %s-%s[%s]" % (m.src_operator_graph_id, srctype.name, srctype.type, m.dst_operator_graph_id, dsttype.name, dsttype.type))
        

    @classmethod
    def get_algo_input_params_without_mapping(cls, params: Dict[str, List[FieldInfo]], type_check: bool = False) -> List[FieldInfo]:
        field_info_map: Dict[str, FieldInfo] = {}
        for field_list in params.values():
            for field in field_list:
                if field.name in field_info_map:
                    if type_check:
                        if field_info_map[field.name].type != field.type:
                            raise ValueError(f"Field name '{field.name}' has different types: '{field_info_map[field.name].type}' and '{field.type}'")
                else:
                    field_info_map[field.name] = field
        return list(field_info_map.values())


    def get_upstream_components_results_types(
        self, upstream_components: List[OperatorComponent], op_dict: Dict[str, Operator]
    ) -> Dict[str, FieldInfo]:
        # note: use op_list for cache to avoid load operator again and again
        # merge all upstream results
        cur_rets: Dict[str, FieldInfo] = {}
        for uc in upstream_components:
            if uc.component_id in op_dict:
                extracted_results = op_dict[uc.component_id].extract_result()
                for key, value in extracted_results.items():
                    cur_rets[f"{uc.component_id}_{key}"] = value
            else:
                raise ValueError(f"Operator info not found for component: {uc}")
        return cur_rets
    
    def get_operator_by_graph_id(self, runnable_graph_id: str) -> Operator:
        # get operator info by graph id
        if runnable_graph_id not in self.operator_infos_by_graph_id:
            raise ValueError(f"Operator info not found for component: {runnable_graph_id}")

        op_info = self.operator_infos_by_graph_id[runnable_graph_id]
        return Operator(op_info)

    def _param_mapping_for_argv(
        self,
        runnable_graph_id: str,
        results: Dict[str, Any],
        argv_input: Any,
        param_mappings: ParameterMappingService,
    ) -> Dict[str, Any]:
        argv = argv_input
        # note: use op_dict for cache to avoid load operator again and again
        op_dict = self.get_all_operators_by_id()

        graph_operator_component_order = self.graph_operator_component_order
        graph_operator_component_order_dict = {
            g.component_graph_id: g for g in graph_operator_component_order
        }
        if runnable_graph_id not in graph_operator_component_order_dict:
            raise ValueError(
                f"current operator info not found for component: {runnable_graph_id}"
            )

        cur_op_name = graph_operator_component_order_dict[
            runnable_graph_id
        ].component_id
        cur_op = op_dict[cur_op_name]

        upstram_components = self.graph.find_upstream_nodes(runnable_graph_id)
        if upstram_components:
            # merge all upstream results
            upstream_return_types = self.get_upstream_components_results_types(
                upstram_components, op_dict
            )
            # cur_params
            current_operator_param_types = cur_op.extract_params()

            upstream_components_graph_ids = [
                u.component_graph_id for u in upstram_components
            ]

            #TODO: refactor
            mappings = param_mappings.get_mappings()
            #logger.debug("mapping count: %d, mapping content: %s", len(mappings), mappings)
            for m in mappings:
                if m.dst_operator_graph_id == runnable_graph_id:
                    if m.src_operator_graph_id not in upstream_components_graph_ids:
                        continue

                    if m.dst_param_name not in current_operator_param_types:
                        raise ValueError(
                            f"Mapping destination param not found for {m.dst_param_name}"
                        )

                    if f"{m.source_operator_id}_{m.src_param_name}" not in upstream_return_types:
                        raise ValueError(
                            f"Mapping source param not found for {m.source_operator_id}_{m.src_param_name}, current upstream types: {upstream_return_types}"
                        )

                    # type check!
                    ret_field = upstream_return_types[f"{m.source_operator_id}_{m.src_param_name}"]
                    src_field = current_operator_param_types[m.dst_param_name]
                    if ret_field.type != src_field.type:
                        raise ValueError(
                            f"Type not match for {m.source_operator_id}_{m.src_param_name} and {m.dst_param_name}, {ret_field.type} != {src_field.type}"
                        )

                    result_op_upstraem = results[m.src_operator_graph_id]
                    if isinstance(result_op_upstraem, dict):
                        val = result_op_upstraem[m.src_param_name]
                    else:
                        val = getattr(result_op_upstraem, m.src_param_name)
                    argv[m.dst_param_name] = val
                    logger.debug(
                        "Mapping %s_%s to %s with value type %s%s",
                        m.source_operator_id,
                        m.src_param_name,
                        m.dst_param_name,
                        type(val).__name__,
                        f" (length={len(val)})" if hasattr(val, "__len__") else "",
                    )
        return argv

    #TODO: use classmethod
    def _set_plugin_to_ctx(self, ctx, runnable_graph_id, plugins_param):
        if not runnable_graph_id in ctx:
            ctx[runnable_graph_id] = {}
        ctx[runnable_graph_id][Operator.INVOKE_PLUGINS_KEY] = plugins_param

    #TODO: use classmethod
    def _set_param_to_ctx(self, ctx, runnable_graph_id, param):
        if not runnable_graph_id in ctx:
            ctx[runnable_graph_id] = {}
        ctx[runnable_graph_id][Operator.INVOKE_PARAM_KEY] = param

    #TODO: use classmethod
    def _set_result_to_ctx(self, ctx, runnable_graph_id: str, result):
        if not runnable_graph_id in ctx:
            ctx[runnable_graph_id] = {}
        ctx[runnable_graph_id][Operator.INVOKE_RESULT_KEY] = result

    def _get_upstream_ctx_results(self, ctx, runnable_graph_id):
        upstram_components = self.graph.find_upstream_nodes(runnable_graph_id)
        if not upstram_components:
            return {}

        results = {
            uc.component_graph_id: ctx[uc.component_graph_id]["result"]
            for uc in upstram_components
        }
        return results

    def mapping_operator_plugins_to_extra_param(self, ctx, runnable: Runnable, argv, operator: Operator):
        invoke_param = self._mapping_operator_plugins_to_extra_param(ctx, runnable.runnable_name, argv, operator)
        if invoke_param:
            runnable.set_extra_param(**invoke_param)

    def _mapping_operator_plugins_to_extra_param(self, ctx, runnable_graph_id: str, argv, operator: Operator):
        #TODO: add unit test
        if not Operator.INVOKE_PLUGINS_KEY in argv:
            return 

        invoke_param: Dict[str, PluginHelper] = {}
        plugin_dict = argv[Operator.INVOKE_PLUGINS_KEY] 
        # filetr that operator needed

        if self.plugin_manager:
            invoke_param = operator.get_operator_argv_plugin_object(self.plugin_manager, plugin_dict)
            # get plugin object
            invoke_param_obj = {k: v.get_plugin_ojbect() for k, v in invoke_param.items()}
            logger.debug("invoke extra param before mapped: %s", invoke_param)
        else:
            logger.warning("No plugin manager set for algorithm, all plugins will be ignored")
        self._set_plugin_to_ctx(ctx, runnable_graph_id, plugin_dict)
        return invoke_param_obj

    
    def mapping_all_plugins_to_extra_param(self, ctx, runnable: Runnable, argv):
        runnable_graph_id = runnable.runnable_name
        if not Operator.INVOKE_PLUGINS_KEY in argv:
            return 

        invoke_param: Dict[str, PluginHelper] = {}
        plugin_dict = argv[Operator.INVOKE_PLUGINS_KEY]
        # filetr that operator needed

        if self.plugin_manager:
            invoke_param = Operator.get_plugin_objects_from_argv(self.plugin_manager, plugin_dict)
            invoke_param_obj = {k: v.get_plugin_ojbect() for k, v in invoke_param.items()}
            logger.debug("invoke extra param before mapped: %s", invoke_param_obj)
        else:
            logger.warning("No plugin manager set for algorithm, all plugins will be ignored")
        
        self._set_plugin_to_ctx(ctx, runnable_graph_id, plugin_dict)

        runnable.set_extra_param(**invoke_param_obj)

    def mapping_param_to_ctx_argv(self, ctx, runnable: Runnable, argv):
        runnable_graph_id = runnable.runnable_name
        self._mapping_param_to_ctx_argv(ctx, runnable_graph_id, argv)
        # set runnable params
        runnable.set_run_param(**argv)

    def _mapping_param_to_ctx_argv(self, ctx, runnable_graph_id: str, argv):
        upstram_components = self.graph.find_upstream_nodes(runnable_graph_id)
        if upstram_components:

            # ARGV MAPPING
            argv_mapped = self._param_mapping_for_argv(
                runnable_graph_id,
                self._get_upstream_ctx_results(ctx, runnable_graph_id),
                argv[Operator.INVOKE_PARAM_KEY],
                self.mappings,
            )

            logger.debug("invoke param before mapped - keys and types: %s", [(k, type(v).__name__ if not isinstance(v, list) else f"list[{len(v)}]") for k, v in argv[Operator.INVOKE_PARAM_KEY].items()])

            logger.debug("invoke param after mapped - keys and types: %s", [(k, type(v).__name__ if not isinstance(v, list) else f"list[{len(v)}]") for k, v in argv_mapped.items()])
            argv[Operator.INVOKE_PARAM_KEY] = argv_mapped
        else:
            logger.debug("No upstream components for %s", runnable_graph_id)

        self._set_param_to_ctx(ctx, runnable_graph_id, argv)

    def _mapping_result_to_ctx(self, ctx, runnable_graph_id: str, result):
        logger.debug("set %s result  to ctx", runnable_graph_id)
        self._set_result_to_ctx(ctx, runnable_graph_id, result)

    def mapping_result_to_ctx(self, ctx, runnable: Runnable, result):
        runnable_graph_id = runnable.runnable_name
        self._mapping_result_to_ctx(ctx, runnable_graph_id, result)
